﻿//This file is part of the Generic Database Accessor Library (GDALib) project
//Copyright (C) 2009 Gentea Alexandru <in.10se@yahoo.com>
//
//This library is free software; you can redistribute it and/or
//modify it under the terms of the GNU Lesser General Public
//License as published by the Free Software Foundation; either
//version 2.1 of the License, or (at your option) any later version.
//
//This library is distributed in the hope that it will be useful,
//but WITHOUT ANY WARRANTY; without even the implied warranty of
//MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
//Lesser General Public License for more details.
//
//You should have received a copy of the GNU Lesser General Public
//License along with this library; if not, write to the Free Software
//Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
using System;
using System.Data.SqlClient;
using System.Data;
using System.Threading;
using System.Text.RegularExpressions;
using System.Collections.Generic;
using System.Text;


namespace GDALib
{
    /// <summary>
    /// Represents the handler function that is called when BulkExec or asynchronous non-queries exceptions arise.
    /// </summary>
    /// <param name="sender">The object where the exception was raised.</param>
    /// <param name="e">The object which encapsulates information about the event.</param>
    public delegate void ErrorEventHandler(object sender, DatabaseAccessorEventArgs e);

    /// <summary>
    /// Represents the function that is called in order to log error messages on exceptions.
    /// </summary>
    /// <param name="message">The text message that describes the exception. This includes the Exception.Message field.</param>
    public delegate void LogExceptionFunction(String message);

    enum BulkExecState
    { 
        Running = 1,
        EndCalled = 2,
        EndNow = -2,
        Stopped = -1
    };

    /// <summary>
    /// Represents a layer on top of ADO.NET which simplifies and allows easier execution of many database operations.
    /// </summary>
    public class DatabaseAccessor
    {
        BulkExecState bulkExecState;
        String connectionString;
        String connectionStringAsync;
        //int maximumRetries;
        int timeout;
        int activeAsyncQueries;
        ManualResetEvent queueReadyFlag;
        ManualResetEvent noAsyncQueriesFlag;
        Queue<object[]> execQueue;
        Thread bulkExecThread;
        static Regex paramRegex;
        const int maxCachedCommands = 10;
        Dictionary<String, CachedSqlCommand> commandCache;
        Exception lastException;
        object syncObj;
        LogExceptionFunction LogException;
        /// <summary>
        /// Occurs when a BulkExec exception or error arises. 
        /// </summary>
        public event ErrorEventHandler BulkExecError;
        /// <summary>
        /// Occurs when an exception or error arises during the execution of an asynchronous command.
        /// </summary>
        public event ErrorEventHandler AsyncExecError;

        /// <summary>
        /// Gets the last exception object that might have arrisen in one of the following methods: 
        /// BeginBulkExec(), ExecuteCachedCommand(), ExecuteNonQuery(), ExecuteReader(), ExecuteScalar.
        /// </summary>
        public Exception LastException
        {
            get { return lastException; }
        }

        /// <summary>
        /// Gets the size of BulkExec's queue. Returns 0 if BulkExec hasn't begun.
        /// </summary>
        public int ExecQueueSize
        {
            get
            {
                if (bulkExecState == BulkExecState.Stopped)
                    return 0;

                return execQueue.Count;
            }
        }

        /// <summary>
        /// Gets the number of active asynchronous queries.
        /// </summary>
        public int ActiveAsyncQueries
        {
            get { return activeAsyncQueries; }
        }

        /// <summary>
        /// Gets or sets the timeout value, in seconds, for all commands executed by the DatabaseAccessor object.
        /// </summary>
        public int Timeout
        {
            get { return timeout; }
            set { timeout = value; }
        }

        /// <summary>
        /// Gets a string array containing the identifiers of the cached commands.
        /// </summary>
        public String[] CachedCommands
        {
            get
            {
                String[] keys = new String[commandCache.Count];
                int i = 0;

                foreach (String k in commandCache.Keys)
                {
                    keys[i] = k;
                    i++;
                }

                return keys;
            }
        }

        /// <summary>
        /// Gets or sets the synchronous connection string. The asynchronous connection string is infered from the input.
        /// NB: Set the connection string only when there's no active BulkExec operation and no asynchronous commands are running.
        /// </summary>
        public String ConnectionString
        {
            get { return connectionString; }
            set
            {
                connectionString = value;

                if (connectionString.EndsWith(";"))
                    connectionStringAsync = value + "Async=true;";
                else
                    connectionStringAsync = value + ";Async=true;";
            }
        }

        /// <summary>
        /// Initializes a new instance of the DatabaseAccessor class.
        /// </summary>
        /// <param name="connectionString">The synchronous connection string.</param>
        /// <param name="LogException">The function that will be called to log exceptions.
        /// If null is provided DatabaseAccessor will use Console.WriteLine by default.</param>
        /// <param name="timeout">The maximum amount of time, in seconds, to wait for a command to complete.</param>
        public DatabaseAccessor(String connectionString, LogExceptionFunction LogException, int timeout)
        {
            this.connectionString = connectionString;
            this.connectionStringAsync = connectionString + (connectionString.EndsWith(";") ? "Async=true;" : ";Async=true;");
            //this.maximumRetries = maximumRetries;
            this.timeout = timeout;
            Initialize(LogException);
        }

        /// <summary>
        /// Initializes a new instance of the DatabaseAccessor class. The connection string will use integrated security for authentication.
        /// </summary>
        /// <param name="server">The server (data source).</param>
        /// <param name="LogException">The function that will be called to log exceptions.
        /// If null is provided DatabaseAccessor will use Console.WriteLine by default.</param>
        public DatabaseAccessor(String server, LogExceptionFunction LogException)
        {
            this.connectionString = "Data Source=" + server + ";Initial Catalog=;Integrated Security=True;Min Pool Size=5;";
            this.connectionStringAsync = connectionString + (connectionString.EndsWith(";") ? "Async=true;" : ";Async=true;");
            Initialize(LogException);
        }

        /// <summary>
        /// Initializes a new instance of the DatabaseAccessor class. The connection string will use integrated security for authentication.
        /// </summary>
        /// <param name="server">The server (data source).</param>
        /// <param name="database">The database name to connect to.</param>
        /// <param name="LogException">The function that will be called to log exceptions.
        /// If null is provided DatabaseAccessor will use Console.WriteLine by default.</param>
        public DatabaseAccessor(String server, String database, LogExceptionFunction LogException)
        {
            this.connectionString = String.Format("Data Source={0};Initial Catalog={1};Integrated Security=True;Min Pool Size=5;", 
                server, database);
            this.connectionStringAsync = connectionString + (connectionString.EndsWith(";") ? "Async=true;" : ";Async=true;");
            Initialize(LogException);
        }

        /// <summary>
        /// Initializes a new instance of the DatabaseAccessor class. 
        /// The connection will authenticate using the supplied username and password.
        /// </summary>
        /// <param name="server">The server (data source).</param>
        /// <param name="database">The database name to connect to.</param>
        /// <param name="user">The username to use for database authentication.</param>
        /// <param name="pass">The corresponding password for the given username.</param>
        /// <param name="LogException">The function that will be called to log exceptions.
        /// If null is provided DatabaseAccessor will use Console.WriteLine by default.</param>
        public DatabaseAccessor(String server, String database, String user, String pass, LogExceptionFunction LogException)
        {
            this.connectionString = String.Format(
                "Data Source={0};Initial Catalog={1};Integrated Security=False;User ID={2};Password={3};Min Pool Size=5;", 
                server, database, user, pass);
            this.connectionStringAsync = connectionString + (connectionString.EndsWith(";") ? "Async=true;" : ";Async=true;");
            Initialize(LogException);
        }

        /// <summary>
        /// Initializes a new instance of the DatabaseAccessor class. 
        /// The connection will authenticate using the supplied username and password.
        /// </summary>
        /// <param name="server">The server (data source).</param>
        /// <param name="user">The username to use for database authentication.</param>
        /// <param name="pass">The corresponding password for the given username.</param>
        /// <param name="LogException">The function that will be called to log exceptions.
        /// If null is provided DatabaseAccessor will use Console.WriteLine by default.</param>
        public DatabaseAccessor(String server, String user, String pass, LogExceptionFunction LogException)
        {
            this.connectionString = String.Format(
                "Data Source={0};Initial Catalog=;Integrated Security=False;User ID={1};Password={2};Min Pool Size=5;",
                server, user, pass);
            this.connectionStringAsync = connectionString + (connectionString.EndsWith(";") ? "Async=true;" : ";Async=true;");
            Initialize(LogException);
        }

        private void Initialize(LogExceptionFunction LogException)
        {
            if (LogException == null)
                this.LogException = new LogExceptionFunction(Console.WriteLine);
            else
                this.LogException = LogException;

            this.activeAsyncQueries = 0;
            this.noAsyncQueriesFlag = new ManualResetEvent(true);
            this.commandCache = new Dictionary<string, CachedSqlCommand>(maxCachedCommands);
            this.syncObj = new object();
            bulkExecState = BulkExecState.Stopped;
        }

        static DatabaseAccessor()
        {
            paramRegex = new Regex("@[a-zA-Z0-9]+", RegexOptions.Compiled);
        }

        private void OnBulkExecError(DatabaseAccessorEventArgs e)
        {
            if (BulkExecError != null)
                BulkExecError(this, e);
        }

        private void OnAsyncExecError(DatabaseAccessorEventArgs e)
        {
            if (AsyncExecError != null)
                AsyncExecError(this, e);
        }

        /// <summary>
        /// Caches a reader, non-query or scalar command in order to allow fast execution.
        /// </summary>
        /// <param name="identifier">The command's unique identifier.</param>
        /// <param name="commandTemplate">The command's template. Parameters must be preceded by the '@' symbol. 
        /// Example: INSERT INTO Sales(SaleId, SaleValue) VALUES(@SaleId, @SaleValue);</param>
        /// <param name="type">The type of the command. This parameter determines the type of output returned by the ExecuteCachedCommand() function. </param>
        /// <returns> Returns true or false depending on the succesful caching of the command.</returns>
        public bool CacheCommand(String identifier, String commandTemplate, SqlCommandType type)
        {
            SqlCommand cmd;

            if (commandCache.Count < maxCachedCommands && identifier.Length > 0 &&
                GetParameterizedCommand(commandTemplate, out cmd))
            {
                commandCache[identifier] = new CachedSqlCommand(cmd, type);
                return true;
            }

            return false;
        }

        /// <summary>
        /// Empties all commands stored in the cache by closing their connections and disposing them.
        /// </summary>
        public void EmptyCache()
        {
            //close the connections, dispose the commands and then empty the dictionary
            foreach (KeyValuePair<String, CachedSqlCommand> cmd in commandCache)
            {
                cmd.Value.command.Connection.Close();
                cmd.Value.command.Dispose();
            }

            commandCache.Clear();
        }

        /// <summary>
        /// Gets the text (template) of a cached command.
        /// </summary>
        /// <param name="identifier">The unique identifier of the command.</param>
        /// <returns></returns>
        public String GetCachedCommandText(String identifier)
        {
            CachedSqlCommand cmd;

            if (commandCache.TryGetValue(identifier, out cmd))
            {
                return cmd.command.CommandText;
            }
            else
                return String.Empty;
        }

        /// <summary>
        /// Deletes a command from the cache by closing its connection and disposing it.
        /// </summary>
        /// <param name="identifier">The unique identifier of the command.</param>
        public void DeleteCachedCommand(String identifier)
        {
            CachedSqlCommand cmd;
            if (commandCache.TryGetValue(identifier, out cmd))
            {
                cmd.command.Connection.Close();
                cmd.command.Dispose();
                commandCache.Remove(identifier);
            }
        }

        /// <summary>
        /// Executes a cached command.
        /// </summary>
        /// <param name="identifier">The unique identifier of the command.</param>
        /// <param name="result">The outputed result object. Its type is of the following: DataTable for reader commands,
        ///  int for non-queries, object for scalars and DataTableCollection for multireaders.</param>
        /// <param name="values">An object array of variable size representing the values for the command template's parameters. 
        /// If the number of values and parameters do not match the command won't be executed.</param>
        /// <returns> Returns true or false depending on the succesful execution of the function. </returns>
        public bool ExecuteCachedCommand(String identifier, out object result, params object[] values)
        {
            CachedSqlCommand cmd;

            //check if the command is in the cache
            if (commandCache.TryGetValue(identifier, out cmd))
            {
                //check to see if the connection is still open
                //if not, try to open it
                if (cmd.command.Connection.State != ConnectionState.Open)
                    try
                    {
                        cmd.command.Connection.Open();
                    }
                    catch (Exception e)
                    {
                        LogException("Failed to open closed database connection. Error: " + e.Message);
                        lastException = e;
                        result = null;
                        return false;
                    }


                //try to set the parameters' values
                if (SetParametersValues(cmd.command, values) == false)
                {
                    LogException("Couldn't set cached command parameters. Values: " + ParametersArrayToString(values));
                    result = null;
                    return false;
                }

                //execute the command
                try
                {
                    switch (cmd.type)
                    {
                        case SqlCommandType.Reader:
                            DataTableCollection dtc = ExecuteReader(cmd.command);
                            if (dtc != null && dtc.Count > 0)
                                result = dtc[0];
                            else
                                result = new DataTable();
                            return true;
                        case SqlCommandType.NonQuery:
                            result = cmd.command.ExecuteNonQuery();
                            return true;
                        case SqlCommandType.Scalar:
                            result = cmd.command.ExecuteScalar();
                            return true;
                        case SqlCommandType.MultiReader:
                            result = ExecuteReader(cmd.command);
                            return true;
                        default:
                            result = null;
                            return false;
                    }
                }
                catch (Exception e)
                {
                    LogException(String.Format("Failed to execute cached command {0}. Type: {1} Text: {2} Values: {3}. Error: {4}",
                        identifier, cmd.type, cmd.command.CommandText, ParametersArrayToString(values), e.Message));
                    lastException = e;
                    result = null;
                    return false;
                }
            }
            else
            {   //if not we don't execute anything
                LogException(String.Format("Command identifier \"{0}\" does not exist in cache", identifier));
                result = null;
                return false;
            }
        }

        private DataTableCollection ExecuteReader(SqlCommand cmd)
        {
            DataSet dataSet = new DataSet();
            SqlDataAdapter dataAdapter;

            //fill the datatable, log it, return it
            dataAdapter = new SqlDataAdapter(cmd);
            dataAdapter.Fill(dataSet);

            if (dataSet.Tables.Count > 0)
                return dataSet.Tables;
            else
                return null;
        }

        /// <summary>
        /// Tests the stored connection string by checking if it's possible to open a connection to the database.
        /// </summary>
        /// <returns>Returns true if an open connection was established and false otherwise. </returns>
        public bool TestConnection()
        {
            using (SqlConnection connection = GetConnection(false))
            {
                return connection != null;
            }
        }

        private SqlConnection GetConnection(bool isAsync)
        {
            try
            {
                SqlConnection connection;
                if (isAsync)
                    connection = new SqlConnection(connectionStringAsync);
                else
                    connection = new SqlConnection(connectionString);

                connection.Open();
                if (connection.State == ConnectionState.Open)
                    return connection;
            }
            catch (Exception e)
            {
                LogException("Failed to establish connection. Error: " + e.Message);
                lastException = e;
            }

            return null;
        }

        /// <summary>
        /// Executes a reader text command and returns a single DataTable. 
        /// If the query didn't complete succesfuly the DataTable will be empty.
        /// </summary>
        /// <param name="sqlStatement">The text of the command to be executed.</param>
        /// <returns>Returns a DataTable, representing the result of the query.</returns>
        public DataTable ExecuteReader(String sqlStatement)
        {
            DataTableCollection dtc = ExecuteMultiReader(sqlStatement);
            if (dtc != null && dtc.Count > 0)
                return dtc[0];
            else
                return new DataTable();
        }

        /// <summary>
        /// Executes a reader text command and returns a DataTableCollection. 
        /// If the query didn't complete succesfuly the result will be null.
        /// </summary>
        /// <param name="sqlStatement">The text of the command to be executed.</param>
        /// <returns>Returns a DataTableCollection, representing the result tables of the query.</returns>
        public DataTableCollection ExecuteMultiReader(String sqlStatement)
        {
            if (sqlStatement.Length == 0)
                return null;

            //create connection
            using (SqlConnection connection = GetConnection(false))
            {
                if (connection == null)
                    return null;

                //create the command
                using (SqlCommand cmd = new SqlCommand(sqlStatement, connection))
                {
                    cmd.CommandTimeout = timeout;

                    try
                    {
                        return ExecuteReader(cmd);
                    }
                    catch (Exception e)
                    {
                        LogException(String.Format("Failed to execute reader: {0}. Error: {1}", sqlStatement, e.Message));
                        lastException = e;
                    }
                }
            }

            return null;
        }

        /// <summary>
        /// Executes a non-query text command and returns the number of rows affected.
        /// </summary>
        /// <param name="sqlStatement">The text of the command to be executed.</param>
        /// <returns>Return the number of rows affected.</returns>
        public int ExecuteNonQuery(String sqlStatement)
        {
            if (sqlStatement.Length == 0)
                return 0;

            int rowsAffected = 0;

            using (SqlConnection connection = GetConnection(false))
            {
                if (connection == null)
                    return 0;

                using (SqlCommand cmd = new SqlCommand(sqlStatement, connection))
                {
                    cmd.CommandTimeout = timeout;

                    try
                    {
                        rowsAffected = cmd.ExecuteNonQuery();
                    }
                    catch (Exception e)
                    {
                        LogException(String.Format("Failed to execute NonQuery: {0}. Error: {1}", sqlStatement, e.Message));
                        lastException = e;
                    }
                }
            }

            return rowsAffected;
        }

        /// <summary>
        /// Executes a non-query text command asynchronously.
        /// </summary>
        /// <param name="sqlStatement">The text of the command to be executed.</param>
        public void ExecuteNonQueryAsync(String sqlStatement)
        {
            if (sqlStatement.Length == 0)
                return;

            //create the connection and command objects
            SqlConnection connectionAsync = GetConnection(true);
            if (connectionAsync == null)
                return;

            SqlCommand cmd = new SqlCommand(sqlStatement, connectionAsync);
            cmd.CommandTimeout = timeout;

            try
            {
                lock (syncObj)
                {
                    cmd.BeginExecuteNonQuery(new AsyncCallback(NonQueryExecuted), cmd);
                    activeAsyncQueries++;
                    noAsyncQueriesFlag.Reset();
                }
            }
            catch (Exception e)
            {
                LogException(String.Format("Failed to begin execution of non query command: {0}. Error: {1}",
                    sqlStatement, e.Message));
                OnAsyncExecError(new DatabaseAccessorEventArgs(e, sqlStatement, null));
            }
        }

        private void NonQueryExecuted(IAsyncResult asyn)
        {
            SqlCommand cmd = (SqlCommand)asyn.AsyncState;

            try
            {
                //finish the execution
                cmd.EndExecuteNonQuery(asyn);
            }
            catch (Exception e)
            {
                LogException(String.Format("Failed to asynchronously execute non-query: {0}. Error: {1}", cmd.CommandText, e.Message));
                OnAsyncExecError(new DatabaseAccessorEventArgs(e, cmd.CommandText, null));
            }
            finally
            {
                lock (syncObj)
                {
                    activeAsyncQueries--;
                    if (activeAsyncQueries == 0)
                        noAsyncQueriesFlag.Set();
                }

                cmd.Connection.Close();
                cmd.Dispose();
            }
        }

        /// <summary>
        /// Executes a reader text command and returns the first column of the first row in the result set.
        /// If the command didn't complete succesfuly the output is null.
        /// </summary>
        /// <param name="sqlStatement">The text of the command to be executed.</param>
        /// <returns>Returns an object, representing the result of the query.</returns>
        public object ExecuteScalar(String sqlStatement)
        {
            if (sqlStatement.Length == 0)
                return null;

            object result = null;

            //create connection and command objects
            using (SqlConnection connection = GetConnection(false))
            {
                if (connection == null)
                    return null;

                using (SqlCommand cmd = new SqlCommand(sqlStatement, connection))
                {
                    cmd.CommandTimeout = timeout;

                    try
                    {
                        //run the command
                        result = cmd.ExecuteScalar();
                    }
                    catch (Exception e)
                    {
                        LogException(String.Format("Failed to execute scalar {0}. Error: {1}", sqlStatement, e.Message));
                        lastException = e;
                    }
                }
            }
            return result;
        }

        /// <summary>
        /// Executes a reader text command and returns the first column of the first row in the result set.
        /// If the command didn't complete succesfuly the output is null.
        /// </summary>
        /// <param name="sqlStatement">The text of the command to be executed.</param>
        /// <returns>Returns an object, representing the result of the query.</returns>
        public Nullable<T> ExecuteScalar<T>(String sqlStatement) where T: struct
        {
            if (sqlStatement.Length == 0)
                return null;

            //create connection and command objects
            using (SqlConnection connection = GetConnection(false))
            {
                if (connection == null)
                    return null;

                using (SqlCommand cmd = new SqlCommand(sqlStatement, connection))
                {
                    cmd.CommandTimeout = timeout;

                    try
                    {
                        //run the command
                        T result = (T)cmd.ExecuteScalar();
                        return result;
                    }
                    catch (Exception e)
                    {
                        LogException(String.Format("Failed to execute scalar {0}. Error: {1}", sqlStatement, e.Message));
                        lastException = e;
                    }
                }
            }

            return null;
        }

        /// <summary>
        /// Begins an asynchronous BulkExec operation wrapped inside a transaction. 
        /// This should be exclusively used for INSERT, UPDATE or DELETE operations.
        /// NB: Choosing a wrong isolation level may lock the database for a long time, depending on the number of rows to be processed.
        /// </summary>
        /// <param name="commandTemplate">The command's template. Parameters must be preceded by the '@' symbol. 
        /// Example: INSERT INTO Sales(SaleId, SaleValue) VALUES(@SaleId, @SaleValue);</param>
        /// <param name="isolationLevel">Specifies the isolation level for the transaction.</param>
        /// <param name="endOnException">When set to false, BulkExec will continue execution although the
        /// previous statements might have been rolled back by the SQL engine, depending on the exception severity. If endOnException
        /// is true, BulkExec stops and explicitly calls rollback when an exception is caught.
        /// </param>
        /// <returns>Returns true if the operation was succesfuly started.</returns>
        public bool BeginBulkExecTrans(String commandTemplate, System.Data.IsolationLevel isolationLevel, bool endOnException)
        {
            return BeginBulkExec(commandTemplate, isolationLevel, endOnException);
        }

        /// <summary>
        /// Begins an asynchronous BulkExec operation wrapped inside a transaction. 
        /// This should be exclusively used for INSERT, UPDATE or DELETE operations.
        /// </summary>
        /// <param name="commandTemplate">The command's template. Parameters must be preceded by the '@' symbol. 
        /// Example: INSERT INTO Sales(SaleId, SaleValue) VALUES(@SaleId, @SaleValue);</param>
        /// <returns>Returns true if the operation was succesfuly started.</returns>
        public bool BeginBulkExec(String commandTemplate)
        {
            return BeginBulkExec(commandTemplate, null, null);
        }

        private bool BeginBulkExec(String commandTemplate, Nullable<IsolationLevel> isolationLevel, Nullable<bool> endOnException)
        {
            //only one bulk insert operation is allowed per instance
            if (bulkExecState != BulkExecState.Stopped)
                return false;

            //preapare the bulk insert thread and sync events
            bulkExecThread = new Thread(new ParameterizedThreadStart(BulkExec));
            if (execQueue == null)
            {
                execQueue = new Queue<object[]>();
                queueReadyFlag = new ManualResetEvent(false);
            }
            else
            {
                execQueue.Clear();
                queueReadyFlag.Reset();
            }

            //bulkExecRunning = true;
            bulkExecState = BulkExecState.Running;

            try
            {
                //start the operation
                bulkExecThread.Name = "BulkExec Thread";
                bulkExecThread.IsBackground = true;
                bulkExecThread.Start(new BulkExecParameter(commandTemplate, isolationLevel, endOnException));
                
                //if everything went ok return true
                if ((bulkExecThread.ThreadState & (ThreadState.Background | ThreadState.WaitSleepJoin | ThreadState.Running)) != 0)
                    return true;
                else
                {
                    //bulkExecRunning = false;
                    bulkExecState = BulkExecState.Stopped;
                    return false;
                }
            }
            catch (Exception e)
            {
                LogException("Failed to start BulkExec thread. Error: " + e.Message);
                lastException = e;
                //bulkExecRunning = false;
                bulkExecState = BulkExecState.Stopped;
                return false;
            }
        }

        /// <summary>
        /// Ends the BulkExec operation.
        /// </summary>
        /// <param name="forceEnd">The forceEnd parameter determines wheter call will be blocking or not.
        /// If forceEnd is set to false, the calling thread will wait for BulkExec to finish processing its queue. 
        /// If forceEnd is set to true, the BulkExec operation ends immediately, unblocking any other thread that was waiting for
        /// BulkExec to finish.</param>
        public void EndBulkExec(bool forceEnd)
        {
            EndBulkExec(forceEnd, -1);
        }

        /// <summary>
        /// Ends the BulkExec operation.
        /// </summary>
        /// <param name="forceEnd">The forceEnd parameter determines wheter call will be blocking or not.
        /// If forceEnd is set to false, the calling thread will wait for BulkExec to finish processing its queue. 
        /// If forceEnd is set to true, the BulkExec operation ends immediately, unblocking any other thread that was waiting for
        /// BulkExec to finish.</param>
        /// <param name="timeout"> The timeout period to wait, in miliseconds, 
        /// for the join operation to complete between the calling thread and the BulkExec thread.</param>
        public void EndBulkExec(bool forceEnd, int timeout)
        {
            if (bulkExecState == BulkExecState.Stopped) 
                return;

            //change the state and signal that the queue is ready in case BulkExec is waiting for data
            lock (execQueue)
            {
                if (forceEnd)
                    bulkExecState = BulkExecState.EndNow;
                else
                    bulkExecState = BulkExecState.EndCalled;

                queueReadyFlag.Set();
            }

            //wait until BulkExec terminates by joining with the BulkExec thread
            if (bulkExecThread.ThreadState != ThreadState.Stopped)
                try
                {
                    if (timeout < 0)
                        bulkExecThread.Join();
                    else
                        bulkExecThread.Join(timeout);
                }
                catch (Exception e)
                {
                    LogException("Failed to join bulk exec thread. Error: " + e.Message);
                }
        }

        /// <summary>
        /// Causes the calling thread to wait until all asynchronous queries are finished. This does not include BulkExec.
        /// </summary>
        public void WaitForAsyncQueries()
        {
            noAsyncQueriesFlag.WaitOne();
        }

        //public void WaitForBulkInsert()
        //{
        //    //if the queue is not empty wait for the bulk insert thread to finish
        //    //while (insertQueue.Count != 0)
        //    //{
        //    //    if (queueEmptyFlag.WaitOne(250))
        //    //        break;
        //    //}
        //    queueEmptyFlag.WaitOne();
            
        //}

        /// <summary>
        /// Enqueues a set of values to be assigned to the BulkExec command for execution.
        /// </summary>
        /// <param name="values">An object array of variable size representing the values for the command template's parameters. 
        /// If the number of values and parameters do not match the command won't be executed.</param>
        public void EnqueForBulkExec(params object[] values)
        {
            lock (execQueue)
            {
                execQueue.Enqueue(values);
                queueReadyFlag.Set();
            }
        }

        private void BulkExec(object paramterObj)
        {
            SqlCommand cmd;
            BulkExecParameter parameter = paramterObj as BulkExecParameter;

            //build the command
            if (GetParameterizedCommand(parameter.CommandTemplate, out cmd) == false)
            {
                OnBulkExecError(new DatabaseAccessorEventArgs(
                    new Exception("Failed to create parameterized command"), null, null));
                bulkExecState = BulkExecState.Stopped;
                return;
            }

            object[] values = null;
            SqlTransaction transaction = null;

            //try to set up a transaction for the bulk exec command
            if (parameter.IsolationLevel.HasValue)
            {
                try
                {
                    transaction = cmd.Connection.BeginTransaction(parameter.IsolationLevel.Value, "BulkExecTrans");
                    cmd.Transaction = transaction;
                }
                catch (Exception e)
                {
                    LogException("Failed to begin bulk exec transaction. Error: " + e.Message);
                }
            }

            while (bulkExecState > 0)
            {
                //if the queue is empty wait
                queueReadyFlag.WaitOne();

                //set the parameters' values and execute the query
                try
                {
                    lock (execQueue)
                    {
                        if (execQueue.Count != 0)
                            values = execQueue.Dequeue();
                        else
                        {   
                            //queueEmptyFlag.Set();
                            if (bulkExecState == BulkExecState.EndCalled)
                            {
                                bulkExecState = BulkExecState.EndNow;
                                break;
                            }
                            else
                            {
                                queueReadyFlag.Reset();
                                continue;
                            }
                        }
                    }
                }
                catch
                {
                    continue;
                }

                if (SetParametersValues(cmd, values))
                {
                    try
                    {
                        cmd.ExecuteNonQuery();
                    }
                    catch (Exception e)
                    {
                        LogException(String.Format("Bulk exec failed for statement: {0} with parameters {1}. Error: {2}",
                            parameter.CommandTemplate, ParametersArrayToString(values), e.Message));

                        OnBulkExecError(new DatabaseAccessorEventArgs(e, parameter.CommandTemplate, values));

                        //if this is wrapped inside a transaction and EndOnException is true
                        if (parameter.EndOnException.HasValue && parameter.EndOnException.Value == true)
                        {
                            if(transaction.Connection != null)
                                try
                                {
                                    transaction.Rollback();
                                }
                                catch (Exception ex)
                                {
                                    LogException("Failed to explicitly rollback BulkExec transaction. Error: " + ex.Message);
                                }

                            //bulkExecRunning = false;
                            bulkExecState = BulkExecState.EndNow;
                            break;
                        }
                    }
                }
            }

            execQueue.Clear();
            //commit the changes if they're part of a transaction
            if (transaction != null && transaction.Connection != null)
            {
                try
                {
                    transaction.Commit();
                }
                catch (Exception e)
                {
                    LogException("Failed to commit bulk exec transaction. Error: " + e.Message);
                }
            }

            //cleanup
            cmd.Connection.Close();
            cmd.Dispose();
            if (transaction != null)
                transaction.Dispose();

            bulkExecState = BulkExecState.Stopped;
        }

        private bool SetParametersValues(SqlCommand cmd, object[] values)
        {
            if (values.Length == cmd.Parameters.Count)
            {
                for (int i = 0; i < values.Length; i++)
                    cmd.Parameters[i].Value = values[i];

                return true;
            }
            else
                return false;
        }

        private String ParametersArrayToString(object[] values)
        {
            StringBuilder output = new StringBuilder();

            for (int i = 0; i < values.Length; i++)
                output.Append((values[i] != DBNull.Value ? values[i] : "NULL") + ",");

            return output.ToString(0, output.Length - 1);
        }

        private bool GetParameterizedCommand(String commandTemplate, out SqlCommand cmd)
        {
            SqlConnection connection = GetConnection(false);

            if (connection == null)
            {
                cmd = null;
                return false;
            }

            cmd = connection.CreateCommand();
            cmd.CommandText = commandTemplate;
            cmd.CommandTimeout = timeout;
            MatchCollection paramsCollection = paramRegex.Matches(cmd.CommandText);

            //create parameters
            foreach (Match p in paramsCollection)
            {
                if (cmd.Parameters.Contains(p.Value))
                    continue;

                SqlParameter sqlParam = cmd.CreateParameter();
                sqlParam.ParameterName = p.Value;
                cmd.Parameters.Add(sqlParam);
            }

            return true;
        }

        /// <summary>
        /// Empty the pools for the synchronous and asynchronous connections.
        /// </summary>
        public void ClearConnectionPool()
        {
            SqlConnection.ClearPool(GetConnection(false));
            SqlConnection.ClearPool(GetConnection(true));
        }

        /// <summary>
        /// Empty the connection pools for every connection, including those not in use by DatabaseAccessor.
        /// </summary>
        public void ClearAllConnectionPools()
        {
            SqlConnection.ClearAllPools();
        }
    }
}
